package com.itbaizhan.api;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;


public class HBaseApiDemo {

    //声明一个静态HBase的连接对象
    private static Connection connection;
    //通过静态代码块创建出来连接对象
    static{
        //创建配置文件对象
        Configuration configuration = HBaseConfiguration.create();
        //设置Zookeeper集群
        configuration.set("hbase.zookeeper.quorum", "node02,node03,node04");
        //创建连接对象
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**创建命名空间
     * @param nameSpace：命名空间的名称
     * @throws IOException
     */
    public static void createNameSpace(String nameSpace) throws IOException {
        //1.获取HBaseAdmin对象：对命名空间、表的CRUD操作使用
        Admin admin = connection.getAdmin();
        //3.创建命名空间的描述器对象
        NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(nameSpace).build();
        //2.创建命名空间
        try{
            admin.createNamespace(namespaceDescriptor);
        }catch (NamespaceExistException nee){
            System.out.println("命名空间："+nameSpace+"已经存在");
        }
        //4.关闭
        admin.close();
        connection.close();
    }

    /**删除命名空间
     * @param nameSpace 命名空间的名称
     */
    public static void deleteNameSpace(String nameSpace){
        //1.参数的合法性检验
        if(nameSpace==null || nameSpace.trim().length()==0){
            System.out.println("命名空间不能为空");
            return;
        }
        //2.声明Admin
        Admin admin = null;
        try{
            //3.创建HBaseAdmin对象
            admin = connection.getAdmin();
            //4.删除命名空间
            admin.deleteNamespace(nameSpace);
        }catch (IOException e){
            e.printStackTrace();
        }finally {
            //5.关闭相关对象
            if(admin!=null){
                try {
                    admin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null){
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    /**判断一张表是否存在
     * @param tableName 表名
     */
    public static boolean tableExist(String tableName){
        //1.声明Admin对象
        Admin admin = null;
        try{
            //2.通过connection对象回去HBaseAdmin对象
            admin = connection.getAdmin();
            //3.将String类型tableName转换为TableName类的对象
            TableName tableNameObj = TableName.valueOf(tableName);
            //4.判断一个表是否存在
            return admin.tableExists(tableNameObj);
        }catch (IOException e){
            e.printStackTrace();
        }finally {
            if(admin!=null){
                try {
                    admin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null){
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return false;
    }

    /**根据指定的表名和列族名称创建表
     * @param tableName:表名称
     * @param cfs:列族名 ...cfs可能传入多个列族
     */
    public static void createTable(String tableName,String ...cfs) throws IOException {
        //1.判断表名是否为空
        if(tableName==null || "".equals(tableName.trim())){
            System.out.println("表名不能为空");
            return;
        }
        //2.验证列族信息的合法性
        if(cfs==null||cfs.length<=0){
            System.out.println("至少需要一个列族");
            return;
        }
        //3.将字符串类型的表名转为TableName对象
        TableName tableNameObj = TableName.valueOf(tableName);
        //4.获取admin对象
        Admin admin = connection.getAdmin();
        //5.判断表是否存在
        if(admin.tableExists(tableNameObj)){
            System.out.println("该表已经存在！");
            admin.close();
            connection.close();
            return;
        }
        //6.创建表描述器对象
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableNameObj);
        //7.遍历cfs，设置列族信息
        for(String cf : cfs){
            //8.使用列族名称，构造ColumnFamilyDescriptorBuilder对象
            ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder =
                    ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf));
            //9.将列族信息添加到表描述器对象
            tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());
        }
        //10.tableDescriptorBuilder构造成TableDescriptor
        TableDescriptor tableDesc = tableDescriptorBuilder.build();
        //11.创建表
        admin.createTable(tableDesc);
        //12.关闭资源
        admin.close();
        connection.close();
    }

    /**判断一个表是否存在（重载tableExist方法）
     * @param tableName 表名
     * @param closeConn 是否关闭connection对象，true表示关闭，false表示不关闭
     * @return true表示存在，false表示不存在
     */
    public static boolean tableExist(String tableName,boolean closeConn){
        //1.声明Admin对象
        Admin admin  = null;
        try {
            //2.通过connection对象回去HBaseAdmin对象
            admin  = connection.getAdmin();
            //3.将String类型tableName转换为TableName类的对象
            TableName tableNameObj = TableName.valueOf(tableName);
            //4.判断一个表是否存在
            return  admin.tableExists(tableNameObj);
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            if(admin!=null){
                try {
                    admin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if(closeConn && connection!=null){
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return false;
    }

    /**删除指定的表
     * @param tableName 表名称
     */
    public static void deleteTable(String tableName) throws IOException {
        //1.校验表名的合法性
        if(tableName==null||"".equals(tableName.trim())){
            System.out.println("表名不能为空");
            return;
        }
        //2.判断表是否存在,重载方法tableExist()
        if(!tableExist(tableName,false)){
            System.out.println("指定的表已经不存在了！");
            return;
        }
        //3.获取Admin
        Admin admin = connection.getAdmin();
        //4.将String类型的tableName转换为TableName类型的对象
        TableName tableNameObj = TableName.valueOf(tableName);
        //5.禁用表
        admin.disableTable(tableNameObj);
        //6.删除表
        admin.deleteTable(tableNameObj);
        //7.关闭资源
        admin.close();
        connection.close();
    }


    /**添加数据
     * @param tableName:表名
     * @param cf：列族名称
     * @param cd：列描述符
     * @param rowKey：数据的唯一表示rowkey
     * @param value：值
     */
    public static void putCellData(String tableName,String cf,String cd, byte[] rowKey,byte[] value) throws IOException {
        //1.校验参数的合法性 tableName,cf,cd  省略
        //2.将String类型的tableName转换为TableName类型的对象
        TableName tableNameObj = TableName.valueOf(tableName);
        //3.通过connection对象获取HTable对象，对数据的CRUD都是通过该对象完成
        Table table = connection.getTable(tableNameObj);
        //4.创建Put对象
        Put put = new Put(rowKey);
        //5.将数据封装到put对象
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes(cd),value);
        //6.添加数据
        table.put(put);
        //7.关闭资源
        table.close();
        connection.close();
    }

    /**get查询一个单元格中的数据
     * @param tableName 表名称
     * @param rowkey 数据的唯一表示
     * @param family 列族名称
     * @param cd 列描述符名称
     */
    public static void getData(String tableName,String rowkey,
                               String family,String cd) throws IOException {
        //1.参数校验省略
        //2.获取HTable对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        //3.创建Get对象
        Get get = new Get(Bytes.toBytes(rowkey));
        //4.指定查询的列族，不指定的话获取rowkey对应这条数据的所有列族下的数据
        get.addFamily(Bytes.toBytes(family));
        //5.指定列描述符,不设置该值则查询对应列族下的所有列的数据
        get.addColumn(Bytes.toBytes(family),Bytes.toBytes(cd));
        //6.执行查询
        Result result = table.get(get);
        //7.解析result对象
        for(Cell cell: result.rawCells()){
            System.out.println("row:"+Bytes.toString(CellUtil.cloneRow(cell))+
                    ",cf:"+Bytes.toString(CellUtil.cloneFamily(cell))+
                    ",cq:"+Bytes.toString(CellUtil.cloneQualifier(cell))+
                    ",value:"+Bytes.toString(CellUtil.cloneValue(cell))
            );
        }
    }

    /**scan扫描表数据
     * @param tableName:表的名称
     * @param startRow:起始的rowkey 包含
     * @param stopRow:结束的rowkey 不包含
     */
    public static void scanTableData(String tableName,String startRow,String stopRow) throws IOException {
        //1.参数的校验省略
        //2.获取table对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        //3.创建Scan对象
        Scan scan = new Scan();
        //4.设置查询的范围
        //包含startRow对应的数据
        scan.setStartRow(Bytes.toBytes(startRow));
        //不包含stopRow
        scan.setStopRow(Bytes.toBytes(stopRow));
        //5.执行查询操作
        ResultScanner resultScanner = table.getScanner(scan);
        //6.解析resultScanner
        for(Result result:resultScanner){
            //7.解析result对象
            for(Cell cell:result.rawCells()){
                System.out.println(
                        "rowkey:"+Bytes.toString(CellUtil.cloneRow(cell))+
                                ",family:"+Bytes.toString(CellUtil.cloneFamily(cell))+
                                ",qualifier:"+Bytes.toString(CellUtil.cloneQualifier(cell))+
                                ",value:"+Bytes.toString(CellUtil.cloneValue(cell))
                );
            }
        }
        //7.关闭资源
        table.close();
        connection.close();
    }
    public static void scanTableData2(String tableName,String startRow,String stopRow) throws IOException {
        //1.参数的校验省略
        //2.获取table对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        //3.创建Scan对象
        Scan scan = new Scan();
        //4.设置查询的范围
        //包含startRow对应的数据
        scan.withStartRow(Bytes.toBytes(startRow));
        //默认是不包含stopRow
        //scan.withStopRow(Bytes.toBytes(stopRow));
        //第二个参数true包含stopRow，false不包含stopRow
        scan.withStopRow(Bytes.toBytes(stopRow),true);
        //5.执行查询操作
        ResultScanner resultScanner = table.getScanner(scan);
        //6.解析resultScanner
        for(Result result:resultScanner){
            //7.解析result对象
            for(Cell cell:result.rawCells()){
                System.out.println(
                        "rowkey:"+Bytes.toString(CellUtil.cloneRow(cell))+
                                ",family:"+Bytes.toString(CellUtil.cloneFamily(cell))+
                                ",qualifier:"+Bytes.toString(CellUtil.cloneQualifier(cell))+
                                ",value:"+Bytes.toString(CellUtil.cloneValue(cell))
                );
            }
        }
        //7.关闭资源
        table.close();
        connection.close();
    }


    /**删除指定行的指定列族下所有的数据
     * @param tableName：表名称
     * @param rowkey:代表对应行数据的唯一标识
     * @param family：列族名称
     * @throws IOException
     */
    public static void deleteData(String tableName,String rowkey,String family) throws IOException {
        //1.参数的合法性校验 省略
        //2.获取HTable对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        //3.创建删除对象
        Delete delete = new Delete(Bytes.toBytes(rowkey));
        //6.指定列族名称
        delete.addFamily(Bytes.toBytes(family));
        //4.执行删除操作
        table.delete(delete);
        //5.关闭资源
        table.close();
        connection.close();
    }
    /**删除指定行的指定列族指定列的数据
     * @param tableName：表名称
     * @param rowkey:代表对应行数据的唯一标识
     * @param family：列族名称
     * @param qualifier：列描述符
     * @throws IOException
     */
    public static void deleteData(String tableName,String rowkey,String family,String qualifier) throws IOException {
        //1.参数的合法性校验 省略
        //2.获取HTable对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        //3.创建删除对象
        Delete delete = new Delete(Bytes.toBytes(rowkey));
        //6.指定列族名称
        delete.addFamily(Bytes.toBytes(family));
        //7.指定列描述符
        delete.addColumn(Bytes.toBytes(family),Bytes.toBytes(qualifier));
        //4.执行删除操作
        table.delete(delete);
        //5.关闭资源
        table.close();
        connection.close();
    }

    //这是一个main方法，是程序的入口：
    public static void main(String[] args) throws IOException {
        //System.out.println(connection);
        //createNameSpace("testdb");
        //deleteNameSpace("testdb");
        //System.out.println(tableExist("psn"));
        //System.out.println(tableExist("psn1"));
        //createTable("psn1","cf1","cf2");
        //deleteTable("psn1");
        //putCellData("psd","cf","name",Bytes.toBytes("rk001"),Bytes.toBytes("tuhao"));
        //putCellData("psd","cf","age",Bytes.toBytes("rk001"),Bytes.toBytes("18"));
        getData("psd","rk001","cf","name");
        //scan测试
        //以下三个添加数据的操作要分别执行
        //putCellData("psn","cf","name",Bytes.toBytes("rk002"),Bytes.toBytes("diaosi"));
        //putCellData("psn","cf","name",Bytes.toBytes("rk003"),Bytes.toBytes("baifumei"));
        //putCellData("psn","cf","name",Bytes.toBytes("rk004"),Bytes.toBytes("aituocuo"));
        //scanTableData("psn","rk001","rk003");
        //重载后的scan测试
        //scanTableData2("psn","rk001","rk003");
        //deleteData("psn","rk001");
        //deleteData("psn","rk002","cf","name");
    }


}
